package com.navinfo.platform.dataanalysis.service.impl;


import com.google.common.collect.Lists;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.utils.DataUtil;
import com.navinfo.platform.common.utils.DateUtils;
import com.navinfo.platform.common.utils.StatisticsDef;
import com.navinfo.platform.common.utils.TidPartitioner;
import com.navinfo.platform.dataanalysis.dto.EventInfoDto;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.sql.Timestamp;
import java.util.*;
import java.util.stream.StreamSupport;

/**
 * 根据事件日统计信息，生成周或月统计信息
 * @author  web 20181203
 */
public class EventDataHandleService implements IDataHandleService {

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        String day = configMap.get("args_1");
        String type = configMap.get("args_2");
        String path = configMap.get(StatisticsDef.WEEK.equals(type) ? "hive.event.week.path" :  "hive.event.month.path");

        Tuple2<String, String> startEndDay = DateUtils.getStartEndDay(day, type);
        path = String.format(path, startEndDay._1);
        System.out.println("进行事件统计信息处理，partDay="+ startEndDay._1+"，path="+ path);

        //RDD转换
        JavaPairRDD<Long, List<EventInfoDto>> convertRDD = convert((Dataset<Row>) obj.get(0), configMap);

        //计算周或月的统计项
        JavaRDD<EventInfoDto> eventInfoRDD = statisticsData(convertRDD, startEndDay._1, spark);

        //保存数据
        long start = System.currentTimeMillis();
        System.out.println("进行事件统计信息处理，保存Parquet文件开始...");
        Dataset<Row> dataFrame = spark.createDataFrame(eventInfoRDD, EventInfoDto.class);
        //dataFrame.select(EventInfoDto.columns()).show();
        dataFrame.select(EventInfoDto.columns()).repartition(1).write().mode(SaveMode.Overwrite).parquet(path);
        System.out.println("进行事件统计信息处理，保存Parquet文件耗时"+ (System.currentTimeMillis()-start));
    }

    /**
     * 将event_info_day数据转成聚合后的{@code <tid,List<EventInfoDto>>}
     * @param eventDataset
     * @param configMap
     */
    private JavaPairRDD<Long, List<EventInfoDto>> convert(Dataset<Row> eventDataset, Map<String, String> configMap) {
        //将Row转换为dto
        JavaPairRDD<Long, EventInfoDto> javaPairRDD = eventDataset.toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, EventInfoDto>) iterator -> {
            Spliterator<Row> spliterator = ((Iterable<Row>) () -> iterator).spliterator();
            return StreamSupport.stream(spliterator, false).map(row -> {
                EventInfoDto eventInfoDto = EventInfoDto.parseRow(row);
                return new Tuple2<>(eventInfoDto.getTid(), eventInfoDto);
            }).iterator();
        });

        //重分区+聚合操作
        int partitionNum = Optional.ofNullable(configMap.get("args_3")).map(u -> Integer.parseInt(u)).orElse(1024);
        return javaPairRDD.partitionBy(new TidPartitioner(partitionNum))
                .aggregateByKey(
                        Lists.<EventInfoDto>newArrayList(),
                        (v1, v2) -> {
                            v1.add(v2);
                            return v1;
                        }, (v1, v2) -> {
                            v1.addAll(v2);
                            return v1;
                        });
    }

    /**
     * 进行周月的统计
     * @param convertRDD
     * @return
     */
    private JavaRDD<EventInfoDto> statisticsData(JavaPairRDD<Long, List<EventInfoDto>> convertRDD, String day , SparkSession spark){
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        final Broadcast<String> partDay = jsc.broadcast(day);

        return convertRDD.mapPartitions(iterator -> {
            return StreamSupport.stream(((Iterable<Tuple2<Long, List<EventInfoDto>>>) () -> iterator).spliterator(), false).map(item -> {
                EventInfoDto eventInfo = new EventInfoDto();
                eventInfo.setTid(item._1);
                List<EventInfoDto> eventInfoDtos = item._2;
                eventInfoDtos.forEach(e -> {
                    eventInfo.setUid(e.getUid());
                    eventInfo.setMileage(eventInfo.getMileage() + e.getMileage());
                    eventInfo.setDuration(eventInfo.getDuration() + e.getDuration());

                    eventInfo.setClutchTimes(eventInfo.getClutchTimes() + e.getClutchTimes());
                    eventInfo.setClutchMileage(eventInfo.getClutchMileage() + e.getClutchMileage());
                    eventInfo.setClutchDuration(eventInfo.getClutchDuration() + e.getClutchDuration());

                    eventInfo.setCsbTimes(eventInfo.getCsbTimes() + e.getCsbTimes());
                    eventInfo.setCsbMileage(eventInfo.getCsbMileage() + e.getCsbMileage());
                    eventInfo.setCsbDuration(eventInfo.getCsbDuration() + e.getCsbDuration());

                    eventInfo.setHsbTimes(eventInfo.getHsbTimes() + e.getHsbTimes());
                    eventInfo.setHsbMileage(eventInfo.getHsbMileage() + e.getHsbMileage());
                    eventInfo.setHsbDuration(eventInfo.getHsbDuration() + e.getHsbDuration());

                    eventInfo.setNgTimes(eventInfo.getNgTimes() + e.getNgTimes());
                    eventInfo.setNgMileage(eventInfo.getNgMileage() + e.getNgMileage());
                    eventInfo.setNgDuration(eventInfo.getNgDuration() + e.getNgDuration());

                    eventInfo.setFoutTimes(eventInfo.getFoutTimes() + e.getFoutTimes());
                    eventInfo.setFoutDuration(eventInfo.getFoutDuration() + e.getFoutDuration());

                    eventInfo.setIdleTimes(eventInfo.getIdleTimes() + e.getIdleTimes());
                    eventInfo.setIdleDuration(eventInfo.getIdleDuration() + e.getIdleDuration());
                });
                int mileage = eventInfo.getMileage();
                int duration = eventInfo.getDuration();

                //设置占比
                eventInfo.setClutch100kmTimes(DataUtil.hundredKmTimes(eventInfo.getClutchTimes(), mileage));
                eventInfo.setClutchMileageProportion(DataUtil.mileageProportion(eventInfo.getClutchMileage(), mileage));
                eventInfo.setClutchDurationProportion(DataUtil.durationProportion(eventInfo.getClutchDuration(), duration));

                eventInfo.setCsb100kmTimes(DataUtil.hundredKmTimes(eventInfo.getCsbTimes(), mileage));
                eventInfo.setCsbMileageProportion(DataUtil.mileageProportion(eventInfo.getCsbMileage(), mileage));
                eventInfo.setCsbDurationProportion(DataUtil.durationProportion(eventInfo.getCsbDuration(), duration));

                eventInfo.setHsb100kmTimes(DataUtil.hundredKmTimes(eventInfo.getHsbTimes(), mileage));
                eventInfo.setHsbMileageProportion(DataUtil.mileageProportion(eventInfo.getHsbMileage(), mileage));
                eventInfo.setHsbDurationProportion(DataUtil.durationProportion(eventInfo.getHsbDuration(), duration));

                eventInfo.setNg100kmTimes(DataUtil.hundredKmTimes(eventInfo.getNgTimes(), mileage));
                eventInfo.setNgMileageProportion(DataUtil.mileageProportion(eventInfo.getNgMileage(), mileage));
                eventInfo.setNgDurationProportion(DataUtil.durationProportion(eventInfo.getNgDuration(), duration));

                eventInfo.setFout100kmTimes(DataUtil.hundredKmTimes(eventInfo.getFoutTimes(), mileage));
                eventInfo.setFoutDurationProportion(DataUtil.durationProportion(eventInfo.getFoutDuration(), duration));

                eventInfo.setIdle100kmTimes(DataUtil.hundredKmTimes(eventInfo.getIdleTimes(), mileage));
                eventInfo.setIdleDurationProportion(DataUtil.durationProportion(eventInfo.getIdleDuration(), duration));

                eventInfo.setUpdateTime(new Timestamp(System.currentTimeMillis()));
                eventInfo.setDay(partDay.value());

                return eventInfo;
            }).iterator();
        });
    }

}
